package com.example.coupon.service.impl;

import com.alibaba.fastjson.JSON;
import com.example.coupon.constant.Constant;
import com.example.coupon.constant.CouponStatus;
import com.example.coupon.dao.CouponDao;
import com.example.coupon.entity.Coupon;
import com.example.coupon.service.IKafkaService;
import com.example.coupon.vo.CouponKafkaMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;
import java.util.Optional;

/**
 * kafka相关的服务接口实现
 * 核心思想：是将cache中的Coupon的状态变化同步到DB中
 * @author wangbo
 * @description:
 * @date 2021/2/1913:41
 */
@Slf4j
@Service
public class KafkaServiceImpl implements IKafkaService {

    @Resource
    private CouponDao couponDao;

    /**
     * 消费优惠券kafka消息
     * @param record {@link ConsumerRecord}
     */
    @Override
    @KafkaListener(topics = {Constant.TOPIC},groupId = "coupon-1")
    public void consumeCouponKafkaMessage(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()){
            Object message = kafkaMessage.get();
            CouponKafkaMessage couponInfo = JSON.parseObject(message.toString(),CouponKafkaMessage.class);
            log.info("Receive CouponKafkaMessage: {}",message.toString());

            CouponStatus status = CouponStatus.of(couponInfo.getStatus());

            switch (status){
                case USABLE:
                    break;
                case EXPIRED:
                    processExpiredCoupons(couponInfo,status);
                    break;
                case USED:
                    processUsedCoupons(couponInfo,status);
                    break;
            }
        }
    }

    /**
     * 处理已使用的用户优惠券
     * @param kafkaMessage
     * @param status
     */
    private void processUsedCoupons(CouponKafkaMessage kafkaMessage,CouponStatus status){
        processCouponByStatus(kafkaMessage,status);
    }

    /**
     * 处理已过期的用户优惠券
     * @param kafkaMessage
     * @param status
     */
    private void processExpiredCoupons(CouponKafkaMessage kafkaMessage,CouponStatus status){
        processCouponByStatus(kafkaMessage,status);
    }

    /**
     * 根据状态处理优惠券信息
     * @param kafkaMessage
     * @param status
     */
    private void processCouponByStatus(CouponKafkaMessage kafkaMessage,CouponStatus status){
        List<Coupon> coupons = couponDao.findAllById(kafkaMessage.getIds());
        if (CollectionUtils.isNotEmpty(coupons) || coupons.size()!=kafkaMessage.getIds().size()){
            log.error("Can Not Find Right Coupon Info: {}",JSON.toJSONString(kafkaMessage));
            return;
        }
        coupons.forEach(c->c.setStatus(status));
        log.info("CouponKafkaMessage Op Coupon Count:{}",couponDao.saveAll(coupons).size());
    }
}
